#!/usr/bin/python3
# -*- coding: utf-8 -*-
"""
消息内容入MYSQL库,日志格式未特殊说明采用最精简的Filebeat采集策略,如果有不同请重新适配格式内容。后面会陆续放出配套的服务端配置
Attributes:
    __version__ (str): "0.0.1"
    __copyright__ (str): "Copyright (C) 2025 z"
    __license__ (str): "MIT"
    __author__ (str): "Z"
"""

from datetime import datetime


# def distribute(message):
#     return (
#         message['senderPlatform'], message['conversationId'], json.dumps(message['atUsers']), message['chatbotCorpId'], message['chatbotUserId'], message['msgId'],message['senderNick'],
#         message['isAdmin'], message['senderStaffId'], datetime.fromtimestamp(message['sessionWebhookExpiredTime'] / 1000), datetime.fromtimestamp(message['createAt'] / 1000),
#         message['senderCorpId'], message['conversationType'], message['senderId'], message['conversationTitle'], message['isInAtList'], message['sessionWebhook'],
#         message['text']['content'], message['robotCode'], message['msgtype'], message['customize_env'], message['source_client_ip'], datetime.now().strftime('%Y-%m-%d %H:%M:%S')
#     )
def convert_iso_to_mysql(iso_time)-> datetime:
    """将ISO 8601时间格式转换为MySQL datetime格式"""
    try:
        dt = datetime.strptime(iso_time, "%Y-%m-%dT%H:%M:%S.%fZ")
        return dt.strftime("%Y-%m-%d %H:%M:%S")
    except ValueError:
        return None
def fluendbit_kubernetes(content):
    return (
        content.get('kubernetes_container_name', ''), content.get('kubernetes_container_hash', ''), content.get('kubernetes_host', ''), content.get('kubernetes_pod_ip', ''), content.get('stream', ''),
        content.get('kubernetes_docker_id', ''), content.get('kubernetes_pod_id', ''), content.get('kubernetes_container_image', ''), content.get('kubernetes_pod_name', ''), content.get('kubernetes_namespace_name', ''),
        content.get('kubernetes_labels_app', ''), content.get('kubernetes_labels_pod_template_hash', ''), content.get('kubernetes_labels_metrics', ''), content.get('log', ''), content.get('time', ''),
        content.get('customize_env', ''), content.get('_index', ''), content.get('source_client_ip', ''), content.get('source_input_method', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def fluendbit_systemd(content):
    return (
        content.get('_PID', ''), content.get('_SYSTEMD_INVOCATION_ID', ''), content.get('_CAP_EFFECTIVE', ''), content.get('PRIORITY', ''), content.get('_SYSTEMD_CGROUP', ''),
        content.get('customize_env', ''), content.get('_GID', ''), content.get('_SELINUX_CONTEXT', ''), content.get('_MACHINE_ID', ''), content.get('_COMM', ''),
        content.get('MESSAGE', ''), content.get('_STREAM_ID', ''), content.get('_BOOT_ID', ''), content.get('_TRANSPORT', ''), content.get('_SYSTEMD_UNIT', ''),
        content.get('_CMDLINE', ''), content.get('SYSLOG_IDENTIFIER', ''), content.get('_SYSTEMD_SLICE', ''), content.get('SYSLOG_FACILITY', ''), content.get('_UID', ''),
        content.get('_HOSTNAME', ''), content.get('_EXE', ''), content.get('_index', ''), content.get('source_client_ip', ''), content.get('source_input_method', ''),
        datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def fluendbit_logs(content):
    return (
        content.get('_index', ''), content.get('log', ''), content.get('customize_env', ''), content.get('source_device', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def get_prometheus(content):
    return (
        content['status'],  content['alertname'], content['hostname'], content['instance'], content['job'], content['monitor'], content['severity'], 
        content['description'], content['summary'], content['value'], content['group'], content['externalURL'],
        content.get('customize_env', ''), content.get('source_client_ip', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def alerts_get(message):
    return (
        convert_iso_to_mysql(message['endsAt']), message.get('fingerprint', ''), convert_iso_to_mysql(message['startsAt']), convert_iso_to_mysql(message['updatedAt']),
        message.get('state', ''), message.get('alertname', ''), message.get('prometheus', ''), message.get('severity', ''),
        message.get('hostname', ''), message.get('instance', ''), message.get('job', ''), message.get('monitor', ''), message.get('value', ''),
        message.get('description', ''), message.get('summary', ''), message.get('customize_env', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def get_elastalert_nginx(message):
    """
    推荐OpenResty nginx,适配日志格式：
    log_format json escape=json '{ "timestamp": "$time_local",' '"server_host":"$server_addr",' '"http_host":"$http_host",' '"nginx_version":"$nginx_version",' '"client_ip":"$remote_addr",' '"remote_user":"$remote_user",' '"request":"$request",' '"request_id":"$request_id" ,' '"request_method":"$request_method",' '"http_user_agent":"$http_user_agent",' '"size":$body_bytes_sent,' '"upstreamhost":"$upstream_addr",' '"url":"$uri",' '"request_body":"$request_body",' '"http_x_forwarded_for":"$http_x_forwarded_for",' '"referer":"$http_referer",' '"scheme":"$scheme",' '"server_protocol":"$server_protocol",' '"status":"$status",' '"cookie":"$http_cookie" ,' '"x-cache":"$upstream_cache_status" ,' '"request_length":"$request_length",'  '"bytes_sent":"$bytes_sent",' '"request_time":"$request_time",' '"upstream_connect_time":"$upstream_connect_time",' '"upstream_header_time":"$upstream_header_time",' '"upstream_response_time":"$upstream_response_time"   }';
    如日志格试不同,请重新适配下面return及sql_mappings段中的get_elastalert_nginx。
    """
    return (
        message.get('request', ''), message.get('referer', ''), message.get('agent_name', ''), message.get('scheme', ''),
        message.get('log_file_path', ''), message.get('log_offset', ''), message.get('upstream_header_time', ''),
        message.get('nginx_version', ''), message.get('customize_env', ''),
        message.get('request_method', ''), message.get('error_data', ''), message.get('error_field', ''),
        message.get('error_message', ''), message.get('http_host', ''), message.get('http_user_agent', ''),
        message.get('remote_user', ''), message.get('upstreamhost', ''), message.get('request_time', ''),
        message.get('host_name', ''), message.get('client_ip', ''), message.get('cookie', ''),
        message.get('server_host', ''), message.get('bytes_sent', ''), message.get('url', ''),
        convert_iso_to_mysql(message.get('@timestamp')), message.get('request_body', ''), message.get('size', ''),
        message.get('upstream_connect_time', ''), message.get('request_length', ''), message.get('http_x_forwarded_for', ''),
        message.get('upstream_response_time', ''), message.get('x-cache', ''), message.get('request_id', ''),
        message.get('status', ''), message.get('server_protocol', ''), message.get('_id', ''),
        message.get('_index', ''), message.get('source_client_ip', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def get_elastalert_nginx_error(message):
    return (
        message.get('agent_name', ''), convert_iso_to_mysql(message.get('@timestamp')), message.get('log_file_path', ''), message.get('log_offset', ''),
        message.get('host_name', ''), message.get('customize_env', ''), message.get('error_data', ''),
        message.get('error_field', ''), message.get('error_message', ''), message.get('message', ''),
        message.get('_id', ''), message.get('_index', ''),  message.get('source_client_ip', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def get_elastalert_syslog(message):
    return (
        convert_iso_to_mysql(message.get('@timestamp')), message.get('log_file_path', ''), message.get('log_offset', ''),
        message.get('host_name', ''), message.get('customize_env', ''), message.get('error_data', ''),
        message.get('_id', ''), message.get('_index', ''), message.get('source_client_ip', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def get_elastalert_java_info(message):
    return (
        message.get('agent_name', ''), convert_iso_to_mysql(message.get('@timestamp')), message.get('log_file_path', ''), message.get('log_offset', ''),
        message.get('host_name', ''), message.get('customize_env', ''), message.get('error_data', ''),
        message.get('error_field', ''), message.get('error_message', ''), message.get('message', ''),
        message.get('_id', ''), message.get('_index', ''), message.get('source_client_ip', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )

def get_elastalert_java_error(message):
    return (
        message.get('agent_name', ''), convert_iso_to_mysql(message.get('@timestamp')), message.get('log_file_path', ''), message.get('log_offset', ''),
        message.get('host_name', ''), message.get('customize_env', ''), message.get('error_data', ''),
        message.get('error_field', ''), message.get('error_message', ''), message.get('message', ''),
        message.get('_id', ''), message.get('_index', ''), message.get('source_client_ip', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def get_elastalert_mysql_slow(message):
    return (
        message.get('agent_name', ''), convert_iso_to_mysql(message.get('@timestamp')), message.get('log_file_path', ''), message.get('log_offset', ''),
        message.get('host_name', ''), message.get('customize_env', ''), message.get('error_data', ''),
        message.get('error_field', ''), message.get('error_message', ''), message.get('message', ''),
        message.get('_id', ''), message.get('_index', ''), message.get('source_client_ip', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def get_elastalert_mysql_error(message):
    return (
        message.get('agent_name', ''), convert_iso_to_mysql(message.get('@timestamp')), message.get('log_file_path', ''), message.get('log_offset', ''),
        message.get('host_name', ''), message.get('customize_env', ''), message.get('error_data', ''),
        message.get('error_field', ''), message.get('error_message', ''), message.get('message', ''),
        message.get('_id', ''), message.get('_index', ''), message.get('source_client_ip', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )
def get_inspection_assertion_exception(message):
    return (
        message.get('title', ''), message.get('domain', ''), message.get('ip', ''), message.get('port', ''),
        message.get('http', ''), message.get('assertion', ''), message.get('customize_env', ''),
        message.get('feature', ''), message.get('story', ''), message.get('tag', ''),message.get('description', ''), message.get('severity', ''),
        message.get('testcase', ''), message.get('issue', ''), message.get('link', ''), datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )


sql_mappings = {
    # 'distribute': {
    #     'insert_query': """
    #     INSERT INTO dingtalk_messages (
    #         sender_platform, conversation_id, at_users, chatbot_corp_id, chatbot_user_id,
    #         msg_id, sender_nick, is_admin, sender_staff_id, session_webhook_expired_time,
    #         create_at, sender_corp_id, conversation_type, sender_id, conversation_title,
    #         is_in_at_list, session_webhook, text_content, robot_code, msgtype, customize_env, source_client_ip, create_time
    #     ) VALUES (
    #         %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
    #     )
    #     """,
    #     'data_func': distribute
    # },
    'fluendbit_kubernetes': {
        'insert_query': """
        INSERT INTO fluendbit_kubernetes (
            kubernetes_container_name, kubernetes_container_hash, kubernetes_host, kubernetes_pod_ip, stream,
            kubernetes_docker_id, kubernetes_pod_id, kubernetes_container_image, kubernetes_pod_name, kubernetes_namespace_name,
            kubernetes_labels_app, kubernetes_labels_pod_template_hash, kubernetes_labels_metrics, log, time,
            customize_env, _index, source_client_ip, source_input_method, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': fluendbit_kubernetes
    },
    'fluendbit_systemd': {
        'insert_query': """
        INSERT INTO fluendbit_systemd (
            _PID, _SYSTEMD_INVOCATION_ID, _CAP_EFFECTIVE, PRIORITY, _SYSTEMD_CGROUP,
            customize_env, _GID, _SELINUX_CONTEXT, _MACHINE_ID, _COMM,
            MESSAGE, _STREAM_ID, _BOOT_ID, _TRANSPORT, _SYSTEMD_UNIT,
            _CMDLINE, SYSLOG_IDENTIFIER, _SYSTEMD_SLICE, SYSLOG_FACILITY, _UID,
            _HOSTNAME, _EXE, _index, source_client_ip, source_input_method,
            create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': fluendbit_systemd
    },
    'fluendbit_logs': {
        'insert_query': """
        INSERT INTO fluendbit_log (
            _index, log, customize_env, source_device, create_time
        ) VALUES (
            %s, %s, %s, %s, %s
        )
        """,
        'data_func': fluendbit_logs
    },
    'get_prometheus': {
        'insert_query': """
        INSERT INTO prometheus_alerts (
            alertstatus, alertname, hostname, instance, job, monitor, severity,
            alertdescription, summary, alertvalue, alertgroup, externalURL, customize_env, source_client_ip, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': get_prometheus
    },
    'alerts_get': {
        'insert_query': """
        INSERT INTO alertmanager (
            endsAt, fingerprint, startsAt, updatedAt, state,
            alertname, prometheus, severity, hostname, instance, job, monitor, value,
            description, summary, customize_env, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': alerts_get
    },
    'get_elastalert_nginx': {
        'insert_query': """
        INSERT logger_nginx (
            request, referer, agent_name, scheme, log_file_path, log_offset, upstream_header_time, nginx_version,
            customize_env, request_method, error_data, error_field, error_message, http_host, http_user_agent, remote_user,
            upstreamhost, request_time, host_name, client_ip, cookie, server_host, bytes_sent, url, timestamp, request_body,
            size, upstream_connect_time, request_length, http_x_forwarded_for, upstream_response_time, x_cache, request_id,
            status, server_protocol, _id, _index, source_client_ip, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': get_elastalert_nginx
    },
    'get_elastalert_nginx_error': {
        'insert_query': """
        INSERT logger_nginx_error (
            agent_name, timestamp, log_file_path, log_offset, host_name, customize_env, error_data, error_field, error_message, message,
            _id, _index, source_client_ip, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': get_elastalert_nginx_error
    },
    'get_elastalert_syslog': {
        'insert_query': """
        INSERT logger_syslog (
            timestamp, log_file_path, log_offset, host_name, customize_env, message, _id, _index, source_client_ip, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': get_elastalert_syslog
    },
    'get_elastalert_java_info': {
        'insert_query': """
        INSERT logger_java_info (
            agent_name, timestamp, log_file_path, log_offset, host_name, customize_env, error_data, error_field, error_message, message,
            _id, _index, source_client_ip, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': get_elastalert_java_info
    },
    'get_elastalert_java_error': {
        'insert_query': """
        INSERT logger_java_error (
            agent_name, timestamp, log_file_path, log_offset, host_name, customize_env, error_data, error_field, error_message, message,
            _id, _index, source_client_ip, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': get_elastalert_java_error
    },
    'get_elastalert_mysql_slow': {
        'insert_query': """
        INSERT logger_mysql_slow (
            agent_name, timestamp, log_file_path, log_offset, host_name, customize_env, error_data, error_field, error_message, message,
            _id, _index, source_client_ip, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': get_elastalert_mysql_slow
    },
    'get_elastalert_mysql_error': {
        'insert_query': """
        INSERT logger_mysql_error (
            agent_name, timestamp, log_file_path, log_offset, host_name, customize_env, error_data, error_field, error_message, message,
            _id, _index, source_client_ip, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': get_elastalert_mysql_error
    },
    'get_inspection_assertion_exception': {
        'insert_query': """
        INSERT inspection_exception (
            title, domain, ip, port, http, assertion, customize_env, feature, story, tag,
            description, severity, testcase, issue, link, create_time
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s
        )
        """,
        'data_func': get_inspection_assertion_exception
    }

}
